package com.cx.support;

import com.cx.config.LettuceRedisAdapter;
import com.cx.exception.CacheConfigException;
import com.cx.exception.CacheException;
import com.cx.property.LettuceProperties;
import com.cx.utils.SerializeAdapter;
import io.lettuce.core.AbstractRedisClient;
import io.lettuce.core.ReadFrom;
import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulConnection;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
import io.lettuce.core.masterslave.MasterSlave;
import io.lettuce.core.masterslave.StatefulRedisMasterSlaveConnection;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection;
import io.lettuce.core.support.AsyncConnectionPoolSupport;
import io.lettuce.core.support.AsyncPool;
import io.lettuce.core.support.BoundedAsyncPool;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.impl.GenericObjectPool;

import java.util.Collections;
import java.util.Map;
import java.util.WeakHashMap;

/**
 * Created on 2017/4/28.
 *
 * @author Alan Shu
 */
public class LettuceConnectionManager {

    private static class LettuceObjects {
        private StatefulConnection<String, Object> connection;
        private AsyncPool asyncPool;
        private GenericObjectPool pool;
        private Object commands;
        private Object asyncCommands;
        private Object reactiveCommands;
    }

    private static final LettuceConnectionManager defaultManager = new LettuceConnectionManager();

    private final LettuceProperties lettuceProperties;

    private Map<AbstractRedisClient, LettuceObjects> map = Collections.synchronizedMap(new WeakHashMap());

    private LettuceConnectionManager() {
        this.lettuceProperties = SerializeAdapter.lettuceProperties();
    }

    public static LettuceConnectionManager defaultManager() {
        return defaultManager;
    }

    private LettuceObjects getLettuceObjectsFromMap(AbstractRedisClient redisClient) {
        LettuceObjects lo = map.get(redisClient);
        if (lo == null) {
            throw new CacheException("LettuceObjects is not initialized");
        }
        return lo;
    }

    public void init(AbstractRedisClient redisClient, StatefulConnection connection, AsyncPool asyncPool) {
        map.computeIfAbsent(redisClient, key -> {
            LettuceObjects lo = new LettuceObjects();
            lo.connection = connection;
            lo.asyncPool = asyncPool;
            return lo;
        });
    }

    public StatefulConnection<String, Object> connection(AbstractRedisClient redisClient) {
        LettuceObjects lo = getLettuceObjectsFromMap(redisClient);
        if (lo.connection == null) {
            if (redisClient instanceof RedisClient) {

                if(null != lettuceProperties.getSentinel()){
                    /**
                     * 卫兵
                     */
                    lo.connection = ((RedisClient) redisClient).connectSentinel(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI());
                } else if (null != lettuceProperties.getCluster()) {
                    /**
                     * 独立主从 | AWS ElastiCache集群
                     */
                    StatefulRedisMasterSlaveConnection<String, Object> connection = MasterSlave.connect((RedisClient) redisClient,
                            LettuceRedisAdapter.objectGZIPCodec(),
                            lettuceProperties.clusterNodes());

                    ReadFrom readFrom = ReadFrom.MASTER_PREFERRED;
                    if (lettuceProperties.getCluster() != null && StringUtils.isNoneBlank(lettuceProperties.getCluster().getReadFromStr())) {
                        readFrom = ReadFrom.valueOf(lettuceProperties.getCluster().getReadFromStr().trim());
                    }

                    connection.setReadFrom(readFrom);

                    lo.connection = connection;
                } else if (lettuceProperties.isOpenPipeline()) {
                    /**
                     * 发布订阅
                     */
                    lo.connection = ((RedisClient) redisClient).connectPubSub(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI());
                } else {
                    /**
                     * 单机
                     */
                    lo.connection = ((RedisClient) redisClient).connect(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI());
                }
            } else if (redisClient instanceof RedisClusterClient) {

                if (lettuceProperties.isOpenPipeline()) {
                    /**
                     * 发布订阅
                     */
                    lo.connection = ((RedisClusterClient) redisClient).connectPubSub(LettuceRedisAdapter.objectGZIPCodec());
                } else {
                    /**
                     * 通用集群
                     */
                    lo.connection = ((RedisClusterClient) redisClient).connect(LettuceRedisAdapter.objectGZIPCodec());
                }
            } else {
                throw new CacheConfigException("type " + redisClient.getClass() + " is not supported");
            }
        }
        return lo.connection;
    }

    // 异步连接池
    public AsyncPool asyncPool(AbstractRedisClient redisClient) {
        LettuceObjects lo = getLettuceObjectsFromMap(redisClient);
        if (lo.asyncPool == null) {
            if (redisClient instanceof RedisClient) {
                if (lettuceProperties.isOpenPipeline()) {
                    /**
                     * 异步发布订阅连接池
                     */
                    lo.asyncPool = AsyncConnectionPoolSupport.createBoundedObjectPool(
                            () -> ((RedisClient) redisClient).connectPubSubAsync(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI()),
                            SerializeAdapter.boundedPoolConfig());
                } else {
                    /**
                     * 异步连接池
                     */
                    lo.asyncPool = AsyncConnectionPoolSupport.createBoundedObjectPool(
                            () -> ((RedisClient) redisClient).connectAsync(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI()),
                            SerializeAdapter.boundedPoolConfig());
                }
            } else if (redisClient instanceof RedisClusterClient) {

                if (lettuceProperties.isOpenPipeline()) {
                    /**
                     * 异步集群发布订阅连接池
                     */
                    lo.asyncPool = AsyncConnectionPoolSupport.createBoundedObjectPool(
                            () -> ((RedisClusterClient) redisClient).connectPubSubAsync(LettuceRedisAdapter.objectGZIPCodec()),
                            SerializeAdapter.boundedPoolConfig());
                } else {
                    /**
                     * 异步集群专用连接池
                     */
                    lo.asyncPool = AsyncConnectionPoolSupport.createBoundedObjectPool(
                            () -> ((RedisClusterClient) redisClient).connectAsync(LettuceRedisAdapter.objectGZIPCodec()),
                            SerializeAdapter.boundedPoolConfig());
                }
            } else {
                throw new CacheConfigException("type " + redisClient.getClass() + " is not supported");
            }
        }

        return lo.asyncPool;
    }

    // 同步连接池
    public GenericObjectPool pool(AbstractRedisClient redisClient) {
        LettuceObjects lo = getLettuceObjectsFromMap(redisClient);
        if (lo.pool == null) {
            if (redisClient instanceof RedisClient) {
                if (lettuceProperties.isOpenPipeline()) {
                    /**
                     * 同步发布订阅连接池
                     */
                    lo.pool = ConnectionPoolSupport.createGenericObjectPool(() ->
                            ((RedisClient) redisClient).connectPubSub(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI()), SerializeAdapter.poolConfig());
                } else {
                    /**
                     * 同步连接池
                     */
                    lo.pool = ConnectionPoolSupport.createGenericObjectPool(() ->
                            ((RedisClient) redisClient).connect(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI()), SerializeAdapter.poolConfig());
                }
            } else if (redisClient instanceof RedisClusterClient) {
                if (lettuceProperties.isOpenPipeline()) {
                    /**
                     * 同步集群发布订阅连接池
                     */
                    lo.pool = ConnectionPoolSupport.createGenericObjectPool(() ->
                            ((RedisClusterClient) redisClient).connectPubSub(LettuceRedisAdapter.objectGZIPCodec()), SerializeAdapter.poolConfig());
                } else {
                    /**
                     * 同步集群专用连接池
                     */
                    lo.pool = ConnectionPoolSupport.createGenericObjectPool(() ->
                            ((RedisClusterClient) redisClient).connect(LettuceRedisAdapter.objectGZIPCodec()), SerializeAdapter.poolConfig());
                }
            } else {
                throw new CacheConfigException("type " + redisClient.getClass() + " is not supported");
            }
        }

        return lo.pool;
    }

    public Object commands(AbstractRedisClient redisClient) {
        connection(redisClient);
        LettuceObjects lo = getLettuceObjectsFromMap(redisClient);
        if (lo.commands == null) {
            if (lo.connection instanceof StatefulRedisConnection) {
                lo.commands = ((StatefulRedisConnection) lo.connection).sync();
            } else if (lo.connection instanceof StatefulRedisClusterConnection) {
                lo.commands = ((StatefulRedisClusterConnection) lo.connection).sync();
            } else if (lo.connection instanceof StatefulRedisSentinelConnection) {
                lo.commands = ((StatefulRedisSentinelConnection) lo.connection).sync();
            } else if (lo.connection instanceof StatefulRedisPubSubConnection) {
                lo.commands = ((StatefulRedisPubSubConnection) lo.connection).sync();
            } else if (lo.connection instanceof StatefulRedisClusterPubSubConnection) {
                lo.commands = ((StatefulRedisClusterPubSubConnection) lo.connection).sync();
            }else {
                throw new CacheConfigException("type " + lo.connection.getClass() + " is not supported");
            }
        }
        return lo.commands;
    }


    public Object asyncCommands(AbstractRedisClient redisClient) {
        connection(redisClient);
        LettuceObjects lo = getLettuceObjectsFromMap(redisClient);
        if (lo.asyncCommands == null) {
            if (lo.connection instanceof StatefulRedisConnection) {
                lo.asyncCommands = ((StatefulRedisConnection) lo.connection).async();
            } else if (lo.connection instanceof StatefulRedisClusterConnection) {
                lo.asyncCommands = ((StatefulRedisClusterConnection) lo.connection).async();
            } else if (lo.connection instanceof StatefulRedisSentinelConnection) {
                lo.asyncCommands = ((StatefulRedisSentinelConnection) lo.connection).async();
            } else if (lo.connection instanceof StatefulRedisPubSubConnection) {
                lo.asyncCommands = ((StatefulRedisPubSubConnection) lo.connection).async();
            } else if (lo.connection instanceof StatefulRedisClusterPubSubConnection) {
                lo.asyncCommands = ((StatefulRedisClusterPubSubConnection) lo.connection).async();
            } else {
                throw new CacheConfigException("type " + lo.connection.getClass() + " is not supported");
            }
        }
        return lo.asyncCommands;
    }

    public Object reactiveCommands(AbstractRedisClient redisClient) {
        connection(redisClient);
        LettuceObjects lo = getLettuceObjectsFromMap(redisClient);
        if (lo.reactiveCommands == null) {
            if (lo.connection instanceof StatefulRedisConnection) {
                lo.reactiveCommands = ((StatefulRedisConnection) lo.connection).reactive();
            } else if (lo.connection instanceof StatefulRedisClusterConnection) {
                lo.reactiveCommands = ((StatefulRedisClusterConnection) lo.connection).reactive();
            } else if (lo.connection instanceof StatefulRedisSentinelConnection) {
                lo.reactiveCommands = ((StatefulRedisSentinelConnection) lo.connection).reactive();
            } else if (lo.connection instanceof StatefulRedisPubSubConnection) {
                lo.reactiveCommands = ((StatefulRedisPubSubConnection) lo.connection).reactive();
            } else if (lo.connection instanceof StatefulRedisClusterPubSubConnection) {
                lo.reactiveCommands = ((StatefulRedisClusterPubSubConnection) lo.connection).reactive();
            } else {
                throw new CacheConfigException("type " + lo.connection.getClass() + " is not supported");
            }
        }
        return lo.reactiveCommands;
    }

    public void removeAndClose(AbstractRedisClient redisClient) {
        LettuceObjects lo = (LettuceObjects) map.remove(redisClient);
        if (lo == null) {
            return;
        }
        /*
        if (lo.commands != null && lo.commands instanceof RedisClusterCommands) {
            ((RedisClusterCommands) lo.commands).close();
        }
        if (lo.asyncCommands != null && lo.asyncCommands instanceof RedisClusterAsyncCommands) {
            ((RedisClusterAsyncCommands) lo.asyncCommands).close();
        }
        if (lo.reactiveCommands != null && lo.reactiveCommands instanceof RedisClusterReactiveCommands) {
            ((RedisClusterReactiveCommands) lo.reactiveCommands).close();
        }
        */
        if (lo.connection != null) {
            lo.connection.close();
        }
        redisClient.shutdown();
    }
}